package com.atguigu.flink.datastreamapi.transform;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/12/13
 *
 *  完全可以替换map和filter
 *
 *     在idea中调试的本地模式，并行度默认是当前机器的cpu的核数。
 */
public class Demo4_FlatMap
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Integer> streamSource = env.fromElements(1, 2, 3, 4, 5, 6, 7);

        //留下奇数，拼接为字符串输出
        streamSource
            .flatMap(new FlatMapFunction<Integer, String>()
        {
            @Override
            public void flatMap(Integer value, Collector<String> out) throws Exception {
                if (value % 2 == 1){
                    out.collect("奇数:"+ value);
                }
            }
        })
        .print();

        env.execute();

    }
}
